package com.ctjsoft.datacollection.core.execute;

import cn.hutool.core.date.DateUtil;
import com.ctjsoft.datacollection.configuration.WebSocketServer;
import com.ctjsoft.datacollection.core.exceptions.MyMessageException;
import com.ctjsoft.datacollection.core.log.KettleLogUtil;
import com.ctjsoft.datacollection.core.repository.RepositoryUtil;
import com.ctjsoft.datacollection.entity.KRepository;
import com.ctjsoft.datacollection.entity.KScriptRecord;
import com.ctjsoft.datacollection.mapper.KRepositoryMapper;
import com.ctjsoft.datacollection.service.KScriptRecordService;
import com.ctjsoft.datacollection.service.impl.KScriptRecordServiceImpl;
import com.ctjsoft.datacollection.util.CollectionUtil;
import com.ctjsoft.datacollection.util.FileUtil;
import com.ctjsoft.datacollection.util.JsonUtil;
import com.ctjsoft.datacollection.util.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.ProgressNullMonitorListener;
import org.pentaho.di.core.Result;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.*;
import org.pentaho.di.core.parameters.UnknownParamException;
import org.pentaho.di.repository.AbstractRepository;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransExecutionConfiguration;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.cluster.TransSplitter;
import org.pentaho.di.www.Carte;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;
import java.util.Map;
import java.util.Random;

/**
 * kettle的ktr脚本执行器
 *
 * @author lyf
 */
@Slf4j
@Component
public class TransExecute {
    static KScriptRecordService scriptRecordService;

    @Autowired
    public KScriptRecordService setKScriptRecordService(KScriptRecordService recordService) {
        if (scriptRecordService == null) {
            scriptRecordService = recordService;
        }
        return scriptRecordService;
    }

    /**
     * ktr执行单元，所有的ktr执行都调用该方法
     *
     * @param tm     ktr元数据
     * @param params ktr需要的命名参数
     * @param args   命令行参数
     */
    private static String executeTrans(TransMeta tm, Map<String, String> params, String[] args,
                                       LogLevel logLevel) throws KettleException {

        /**
         * 同一脚本，同时运行时，日志会串。
         * 原因：
         * LogChannelID是根据transName和stepName来获取的，便于处理
         * 在脚本和步骤名称中加入随机数，保证名称不冲突问题，计算出来的logChannelId不会出现重复的情况
         */
        Random random = new Random();
        int i = random.nextInt(1000);
        // 通过元数据获取ktr的实例
        Trans trans = new Trans(tm);
        LogChannel logChannel = new LogChannel(trans.getName() + i + System.currentTimeMillis());
        trans.setLog(logChannel);
        if (logLevel != null) {
            trans.setLogLevel(logLevel);
        }

        // 传入ktr需要的变量
        if (CollectionUtil.isNotEmpty(params)) {
            log.info("传入kettle的参数：{}", JsonUtil.toJsonString(params));
            params.forEach((String k, Object v) -> {
                try {
                    trans.setParameterValue(k, v.toString());
                } catch (UnknownParamException e) {
                    e.printStackTrace();
                }
            });
        }
        // 开始执行ktr，该方法以传入命令行参数args

//        trans.execute(args);


        KScriptRecord one = scriptRecordService.getById(params.get("recordId"));

        one.setLogChannelId(trans.getLogChannelId());
        scriptRecordService.saveOrUpdate(one);
        // 线程等待，直到ktr执行完成
        trans.waitUntilFinished();

        // 执行完成后获取日志
        String logText1 = KettleLogUtil.getLogText(trans.getLogChannelId(), true,
                                                   trans.getLogDate().getTime());

        // 判断执行过程中是否有错误, 有错误就抛出错误日志
        if (trans.getErrors() > 0) {
            throw new KettleException(logText1);
        }
        // 没有错误就返回正常执行日志
        return logText1;
    }

    /**
     * 运行单个ktr
     *
     * @param fullPathName ktr全路径名
     * @param params       ktr需要的命名参数
     * @param logLevel     日志级别
     */
    public static String run(String fullPathName, Map<String, String> params,
                             LogLevel logLevel) throws KettleException {
        // 通过ktr全路径名获取ktr元数据
        TransMeta tm = new TransMeta(FileUtil.replaceSeparator(fullPathName));
        // 开始执行ktr
        return executeTrans(tm, params, null, logLevel);
    }

    /**
     * 获取资源库
     *
     * @param transRepositoryId 资源库id
     * @return {@link AbstractRepository}
     */
    private static AbstractRepository getAbstractRepository(String transRepositoryId) {
        AbstractRepository repository = RepositoryUtil.getRepository(transRepositoryId);
        KRepositoryMapper repositoryMapper = SpringContextUtil.getBean(KRepositoryMapper.class);
        KRepository kRepository = repositoryMapper.selectById(transRepositoryId);
        if (repository == null || !repository.isConnected()) {
            //Optional<Repository> optionalRepository = repRepository.findById(transRepositoryId);
            if (kRepository == null) {
                throw new MyMessageException("资源库不存在");
            }
            // 连接资源库
            repository = RepositoryUtil.connection(kRepository);
        }
        return repository;
    }

    /**
     * 运行资源库中的ktr
     *
     * @param repositoryId 资源库对象
     * @param transPath    ktr所在路径
     * @param transName    ktr名称
     * @param versionLabel 版本号，传入null表示执行最新的ktr
     * @param params       ktr需要的命名参数
     * @param logLevel     日志级别
     */
    public static String run(String repositoryId, String transPath, String transName, String versionLabel,
                             Map<String, String> params, LogLevel logLevel) throws KettleException {


        AbstractRepository repository = getAbstractRepository(repositoryId);
        // 根据相对目录地址获取ktr所在目录信息
        RepositoryDirectoryInterface rdi = repository
                .loadRepositoryDirectoryTree()
                .findDirectory(FileUtil.getParentPath(transPath));
        // 在指定资源库的目录下找到要执行的转换
        TransMeta tm = repository.loadTransformation(transName, rdi, new ProgressNullMonitorListener(), true,
                                                     versionLabel);
        // 开始执行ktr
        return executeTrans(tm, params, null, logLevel);
    }
}
